package com.doit.shark.dsp.test

import java.io.{BufferedWriter, FileWriter}
import java.util.Random

import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

/**
  * Created by hunter.coder 涛哥  
  * 2019/4/9 17:18
  * 交流qq:657270652
  * Version: 1.0
  * 更多学习资料：https://blog.csdn.net/coderblack/
  *
  * Description: spark graphx 性能测试
  * 机器配置： 32G ，i7 4核8线程  M.2固态硬盘（900M/S 读写速度）   3台虚拟机运行
  * 资源配置： 3个executor ，4g内存/executor， 1core
  * 数据条数： 10万条    40万个点，30万条边      --》  25719子图    54s
  *          100万条   400万个点，300万条边    --》  250950子图   133s
  *          1000万条  4000万个点，3000万条边  --》  内存消耗暴增，虚拟机集群运行失败
  * id个数：  每条4个id
  *
  **/
object GraphxBenchMark {

  def main(args: Array[String]): Unit = {

    if(args(0).equals("1")) {
      genData(args(1))
    }else {
      calc(args(1))
    }

  }

  /**
    * 生成测试数据
    */
  def genData(path:String): Unit = {


    val bw = new BufferedWriter(new FileWriter(path))
    val rd = new Random()

    for (i <- 1 to 10000000) {
      val a = rd.nextInt(100000000)
      val b = rd.nextInt(100000000)
      val c = rd.nextInt(100000000)
      val d = rd.nextInt(100000000)

      bw.write(s"$i,$a,$b,$c,$d")
      bw.newLine()
    }
    bw.close()

  }

  /**
    * 计算连通子图
    */
  def calc(path:String): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession.builder()
      /*.config("spark.executor.memory","8g")*/
      /*.master("local[*]")*/
      .appName("people")
      .getOrCreate()

    val start = System.currentTimeMillis()


    val rdd = spark.read.textFile(path).rdd

    // 构造点集合（RDD[(Long,A)]
    val verticesRDD: RDD[(Long, String)] = rdd.flatMap(line => {

      var list = List[(Long, String)]()
      val split = line.split(",")
      val a = split(1)
      val b = split(2)
      val c = split(3)
      val d = split(4)
      (a.hashCode.toLong, line) :: (b.hashCode.toLong, line) :: (c.hashCode.toLong, line) :: (d.hashCode.toLong, line) :: Nil

    })


    // 构造边集合(RDD[Edge[A]])
    // Edge(srcId:Long,dstId:Long,data:A)
    val edgeRDD: RDD[Edge[String]] = rdd.flatMap(line => {
      val split = line.split(",")
      val a = split(1)
      var list = List[(Edge[String])]()
      for (i <- 2 until split.size) {
        list :+= Edge(a.hashCode.toLong, split(i).hashCode.toLong, "")
      }
      list
    })


    // 利用点集合，边集合，构造一张“图“
    val graph = Graph(verticesRDD, edgeRDD)

    // 调用图的api，获取图中所有的连通子图
    // 返回的连通子图结果点集合：  里面还是之前的图中所有的点（id,属性），每个点的属性=>是这个点所属的连通子图中的最小点id
    val vertices: VertexRDD[VertexId] = graph.connectedComponents().vertices


    println(vertices.map(_._2).distinct().count())

    val end = System.currentTimeMillis()
    println("耗时： " + (end-start)/1000)

    spark.close()

  }


}
